#include "tcp_bridge.h"
// 在多核的机器上，线程池中的数量最好等于核数
serviceThreadPool::serviceThreadPool(int pool_size) : m_io_service_pos(0)
{
    m_poolSize = pool_size > 0 ? pool_size : 1;
    for (int i = 0; i < m_poolSize; i++)
    {
        std::shared_ptr<boost::asio::io_service> io_service(new boost::asio::io_service);
        std::shared_ptr<boost::asio::io_service::work> work(new boost::asio::io_service::work(*io_service));
        m_ioServiceList.push_back(io_service);
        m_workList.push_back(work);
    }
}
void serviceThreadPool::startPool()
{
    for (int i = 0; i < m_poolSize; i++)
    {

        std::thread runThread([=]
                              { m_ioServiceList[i]->run(); });
        runThread.detach();
    }
}
void serviceThreadPool::stop()
{
    for (int i = 0; i < m_poolSize; i++)
    {
        m_ioServiceList[i]->stop();
    }
}
boost::asio::io_service &serviceThreadPool::get_io_service()
{
    // 通过这个函数分配io_service
    boost::asio::io_service &io_service = *m_ioServiceList[m_io_service_pos++];
    if (m_io_service_pos == m_poolSize - 1)
        m_io_service_pos = 0;
    return io_service;
}

/*------------------------------------------------------------------------------------*/
/*------------------------------------------------------------------------------------*/
/*------------------------------------------------------------------------------------*/

rwhandler::rwhandler(boost::asio::io_service &ios) : m_socket(ios)
{
    m_socket_is_valid = true;
    m_getData = new unsigned char[1024];
    RCLCPP_INFO(rclcpp::get_logger("AgvTcpBridge"), "rwhandler create \n");
}
rwhandler::~rwhandler()
{
    delete[] m_getData;
    RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " rwhandler distroy \n");
}
tcp::socket &rwhandler::getSock()
{
    return m_socket;
}
void rwhandler::read()
{
    m_socket.async_read_some(boost::asio::buffer(m_buffer),
        [=](const boost::system::error_code &error, std::size_t bytes_transferred)
    {
        int clientId = this->getClientID();
             // auto clientEndpoint = this->getSock().remote_endpoint();
             // std::string ip_port = clientEndpoint.address().to_string() + ":" + std::to_string(clientEndpoint.port());
             // RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "async_read from client " << clientId << "-" << ip_port << "\n");
        if (error)
        {
            RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " read error! \n");
            handleError(error);
        }
        else
        {
            if (bytes_transferred < RECV_BUFFER_SIZE)
            {
                // 接收个数不足RECV_BUFFER_SIZE时，设置终止符号
                m_buffer[bytes_transferred] = '\0';
            }
            // 将数据通过回调函数或者信号与槽传递回应用层
            m_callbackRecv(clientId, std::string(m_buffer, bytes_transferred));
            this->read();
        }
    });
}
void rwhandler::write(const std::string &message)
{
    m_socket.async_write_some(boost::asio::buffer(message),
        [=](const boost::system::error_code &error, std::size_t bytes_transferred)
    {
        if (error)
        {
            RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " write error! \n");
            handleError(error);
        }
        else
        {
            // RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "writer buffer = " << message.c_str() << ",size=" << bytes_transferred << "\n");
        }
    });
}

void rwhandler::setClientID(int id)
{
    m_clientId = id;
}
int rwhandler::getClientID()
{
    return m_clientId;
}
// 应用层，也就是server中利用SetCallBackError设置执行异常回调的函数
template <typename F>
void rwhandler::SetCallBackError(F f)
{
    m_callbackError = f;
}
template <typename F>
void rwhandler::SetCallBackRecv(F f)
{
    m_callbackRecv = f;
}

void rwhandler::handleError(const boost::system::error_code &ec)
{
    RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " error! code is:" << ec.value() << ec.message() << "\n");

    // 异常码为2表示异步任务等待时远端关闭套接字
    // if (ec.value() == 2)
    if (m_socket_is_valid)
    {
        // 标记套接字无效，防止重复关闭删除调用空指针
        m_socket_is_valid = false;
        // 关闭套接字
        RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " close socket\n");
        boost::system::error_code ec2;
        m_socket.shutdown(tcp::socket::shutdown_send, ec2);
        m_socket.close(ec2);
        if (m_callbackError)
            m_callbackError(m_clientId);
    }
    else
    {
        RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client " << m_clientId << " socket has been processed \n");
    }
}

/*------------------------------------------------------------------------------------*/
/*------------------------------------------------------------------------------------*/
/*------------------------------------------------------------------------------------*/

server::server(boost::asio::io_service &ios, short port, int serPoolSize) : m_ios(ios),
                                                                            m_acceptor(ios, tcp::endpoint(tcp::v4(), port)),
                                                                            m_rwIdList(MAX_CLINET_NUM),
                                                                            m_servicePool(serPoolSize)

{
    int current = 0;
    // 初始化m_rwIdList，产生MAX_CLINET_NUM数量的ID
    std::generate_n(m_rwIdList.begin(), MAX_CLINET_NUM, [&current]
                    { return ++current; });
    m_servicePool.startPool(); // 开启线程池
    accept();
}
server::~server()
{
    m_servicePool.stop();
    RCLCPP_WARN(rclcpp::get_logger("AgvTcpBridge"), "server distroy! \n");
}
void server::setRecvDataProcessor(std::function<void(int, std::string)> processor)
{
    m_vRecvDataProcessor = processor;
}
void server::setConnChangedProcessor(std::function<void(int, bool)> processor)
{
    m_vConnChangedProcessor = processor;
}

void server::sendData(const std::vector<int> &ids, const std::string &buffer)
{
    // 发送数据
    for (auto &id : ids)
    {
        if (m_rwMap.find(id) != m_rwMap.end())
        {
            m_rwMap.at(id)->write(buffer);
        }
    }
}

void server::accept()
{
    // 这里不用智能指针，使用rwhandler rw;就会报“尝试引用已删除对象”的错误
    // 智能指针在离开作用于时应该也是会析构，为什么不会报错呢？
    // 是因为在async_accept的lambda中将rw当作形参传入，这时候rw的引用计数就加1了，因此不会析构
    // 将智能指针放入容器后，智能指引用计数就会加1
    std::shared_ptr<rwhandler> rw = std::make_shared<rwhandler>(m_servicePool.get_io_service());
    rw->setClientID(m_rwIdList.front()); // 设置client ID
    m_rwIdList.pop_front();

    // 设置异常回调函数
    rw->SetCallBackError([this](int id)
                         { reSetRwID(id); });

    rw->SetCallBackRecv([this](int id, std::string buffer)
                        { recvDataCallback(id, buffer); });

    RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "start accept,id:" << rw->getClientID() << ",thread:" << std::this_thread::get_id() << " \n");

    // 这里捕获列表，必须捕获rw,否则智能指针就会自动析构
    m_acceptor.async_accept(rw->getSock(), [this, rw](const boost::system::error_code &error)
    {
			if (error)
			{
				RCLCPP_WARN_STREAM(rclcpp::get_logger("AgvTcpBridge"), "acceptor error:"<< error.value()<< error.message()<<"\n");
				return;
			}
			
            int new_id = rw->getClientID();
			//m_rwList.push_back(rw); //果然，这里在注释掉push_back操作后，智能指针随后就析构了
			{
				std::unique_lock<std::mutex> lock(this->m_lock);
				m_rwMap.insert(std::make_pair(new_id, rw));
			}
			m_rwMap.insert(std::make_pair(new_id, rw));
            m_vConnChangedProcessor(new_id,true);

            auto clientEndpoint = rw->getSock().remote_endpoint();
            std::string ip_port = clientEndpoint.address().to_string()+":"+std::to_string(clientEndpoint.port());
			RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client "<<new_id<<"-"<<ip_port<<" coming in,map size: "<<m_rwMap.size()<<"\n");
			RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "client have acceptor thread id is: "<<std::this_thread::get_id()<<"\n");

			rw->read();
			accept(); });
}
void server::reSetRwID(int id)
{
    std::unique_lock<std::mutex> lock(this->m_lock);

    auto it = m_rwMap.find(id);
    if (it == m_rwMap.end())
    {
        return;
    }

    m_vConnChangedProcessor(id, false);
    m_rwMap.erase(it);

    // 将id重新放入m_rwIdList中，确保添加到最后一个，且无错误重复添加
    auto id_index = std::find(m_rwIdList.begin(), m_rwIdList.end(), id);
    if (id_index != m_rwIdList.end())
    {
        m_rwIdList.erase(id_index);
    }
    m_rwIdList.push_back(id);
    RCLCPP_INFO_STREAM(rclcpp::get_logger("AgvTcpBridge"), "cliend " << id << " exiting,map size: " << m_rwMap.size() << "\n");
}
size_t server::getClientNum()
{
    return m_rwMap.size();
}

void server::recvDataCallback(int id, std::string buffer)
{
    m_vRecvDataProcessor(id, buffer);
}
